package cn._51doit.flink.day10;


import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.util.concurrent.TimeUnit;

/**
 * 将数据使用Flink的FileSink写入到HDFS中（以文本的方式）
 */
public class FileSinkDemo1 {

    public static void main(String[] args) throws Exception {

        System.setProperty("HADOOP_USER_NAME", "root");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60000);
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        String outputPath = "hdfs://node-1.51doit.cn:9000/out28";
        final FileSink<String> sink = FileSink
                .forRowFormat(new Path(outputPath), new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                //如果有数据写入，30秒滚动生成一个文件
                                .withRolloverInterval(TimeUnit.SECONDS.toMillis(30))
                                //如果没有数据写入，10滚动生成一个文件
                                .withInactivityInterval(TimeUnit.SECONDS.toMillis(10))
                                .withMaxPartSize(1024 * 1024 * 1024) //1G按照文件大小滚动生成文件
                                .build())
                .build();

        lines.sinkTo(sink);

        env.execute();

    }
}
